package com.iot08.api.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.iot08.api.queue.LimitedSingleQueue;
import com.iot08.api.service.supplement.TCPService;
import com.iot08.api.utils.TimeUtils;
import com.iot08.common.constant.ConstKafka;
import com.iot08.common.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;


/**
 * @version 1.0
 * @Description: kafka消费者
 * @time: 2020/9/21 14:26
 * @author: Rock
 */
@Slf4j
@Component
public class KafkaConsumer {

    @Resource
    private TCPService tcpService;

    @Value("${useWeiZhiLocateBL_LBS}")
    private String useWeiZhiLocateBL_LBS;

    @Value("${useWeiZhiLocateBL_ordinary}")
    private String useWeiZhiLocateBL_ordinary;

    //不消费kafka消息,用于消息堵塞时快速消费消息的操作 (目前只用于普通主题)
    @Value("${notConsumerKafkaMsg:0}")
    private String notConsumerKafkaMsg;

    //不消费kafka消息,用于消息堵塞时快速消费消息的操作 (用于VIP主题)
    @Value("${notConsumerKafkaMsg_VIP:0}")
    private String notConsumerKafkaMsg_VIP;

    //不消费kafka消息,用于消息堵塞时快速消费消息的操作 (用于LBS主题)
    @Value("${notConsumerKafkaMsg_LBS:0}")
    private String notConsumerKafkaMsg_LBS;

    @Value("${isCloseTcpPushConsumer:0}")
    private Integer isCloseTcpPushConsumer;
    
    @Value("${isCloseTcpPushVipConsumer:0}")
    private Integer isCloseTcpPushVipConsumer;

    @Value("${isCloseTcpPushLbsConsumer:0}")
    private Integer isCloseTcpPushLbsConsumer;

    @Value("${isCloseTcpPushWechatVedioConsumer:0}")
    private Integer isCloseTcpPushWechatVedioConsumer;
    
    /*
    * 这里注意,分布式部署时, 下面3个方法 listen listenLBS listenVIP 需要按需开启和关闭
    * */

    /**
     * @param messages
     * @time: 2021/1/28 15:25
     * @author: Rock
     * @version 1.0
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.group.id}", topics = "${spring.kafka.consumer.topic.name}", concurrency = "${spring.kafka.consumer.concurrency}")
    public void listen(List<ConsumerRecord<String, String>> messages){
        if(Integer.valueOf(1).equals(isCloseTcpPushConsumer)) {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        log.info("topic为jianyou_tongxin_TCPPUSH的kafka消息队列接收到的消息为:{}",message);
                    }
                }
            }
        } else {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        String imei = message.key();
                        String value = message.value().replace("\r", "").replace("\n", "").replace("\ufeff", "");
                        log.info("topic={}. key={}. value={}. partition={} notConsumerKafkaMsg={}", message.topic(),
                                imei, value, message.partition(), notConsumerKafkaMsg);
                        try{
                            JSONObject data = JSON.parseObject(value);
                            if(data != null && StringUtil.isNotBlank(imei)){
                                String kafkaDataType = String.valueOf(data.get(ConstKafka.TYPE));
                                //kafka消费模式: 0:普通 1:LBS 2:VIP
                                data.put("kafkaModelType", 0);
                                consumerKafkaData(kafkaDataType, data, imei);
                                /*tcpService.useWeiZhiLocateBL = Double.valueOf(useWeiZhiLocateBL_ordinary);
                                tcpService.cmdTempFunc(imei, data);*/
                            }
                        } catch (Exception e){
                            log.error(String.format("普通消费主题  Exception = %s. imei = %s. message = %s", e.getMessage(), imei, message), e);
                        }
                    } else {
                        log.error("listen message is null.");
                    }
                }
            } else {
                log.error("listen messages is null.");
            }
        }
    }

    /**
     * @param messages
     * @time: 2021/1/28 15:25
     * @author: Rock
     * @version 1.0
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.group.id-vip}", topics = "${spring.kafka.consumer.topic.name-vip}", concurrency = "${spring.kafka.consumer.concurrency-vip}")
    public void listenVip(List<ConsumerRecord<String, String>> messages){
       if(Integer.valueOf(1).equals(isCloseTcpPushVipConsumer)) {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        log.info("topic为jianyou_tongxin_TCPPUSH_VIP的kafka消息队列接收到的消息为:{}",message);
                    }
                }
            }
        } else {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        String imei = message.key();
                        String value = message.value().replace("\r", "").replace("\n", "")
                                .replace("\ufeff", "");
                        log.info("topic={}. imei={} value={}. partition={}", message.topic(), imei, value, message.partition());
                        try{
                            JSONObject data = JSON.parseObject(value);
                            if(data != null && StringUtil.isNotBlank(imei)){
                                tcpService.useWeiZhiLocateBL = Double.valueOf(useWeiZhiLocateBL_LBS);
                                //kafka消费模式: 0:普通 1:LBS 2:VIP
                                data.put("kafkaModelType", 2);
                                tcpService.cmdTempFunc(imei, data);
                            }
                        } catch (Exception e){
                            log.error(String.format("VIP消费主题 Exception = %s. imei = %s. message = %s", e.getMessage(), imei, message), e);
                        }
                    } else {
                        log.error("listenVip message is null.");
                    }
                }
            } else {
                log.error("listenVip messages is null.");
            }
        }
    }

    /**
     * @param messages
     * @time: 2021/1/28 15:25
     * @author: Rock
     * @version 1.0
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.group.id-lbs}", topics = "${spring.kafka.consumer.topic.name-lbs}", concurrency = "${spring.kafka.consumer.concurrency-lbs}")
    public void listenLbs(List<ConsumerRecord<String, String>> messages){
        if(Integer.valueOf(1).equals(isCloseTcpPushLbsConsumer)) {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        log.info("topic为jianyou_tongxin_TCPPUSH_LBS的kafka消息队列接收到的消息为:{}",message);
                    }
                }
            }
        } else {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        String imei = message.key();
                        String value = message.value().replace("\r", "").replace("\n", "")
                                .replace("\ufeff", "");
                        log.info("topic={}. imei={}. value={}. partition={}", message.topic(), imei, value, message.partition());
                        try{
                            JSONObject data = JSON.parseObject(value);
                            if(data != null && StringUtil.isNotBlank(imei)){
                                String kafkaDataType = String.valueOf(data.get(ConstKafka.TYPE));
                                //kafka消费模式: 0:普通 1:LBS 2:VIP
                                data.put("kafkaModelType", 1);
                                consumerKafkaData(kafkaDataType, data, imei);
                            }
                        } catch (Exception e){
                            log.error(String.format("LBS消费主r题 Exception = %s. imei = %s. message = %s", e.getMessage(), imei, message), e);
                        }
                    } else {
                        log.error("listenLbs message is null.");
                    }
                }
            } else {
                log.error("listenLbs messages is null.");
            }
        }
    }

    /**
     * @description 消费kafka数据，对带定位的消息
     * @author cwj
     * @param [kafkaDataType, data, imei]
     * @return void
     * @date @time 2025/6/4 16:38
     **/
    private void consumerKafkaData(String kafkaDataType, JSONObject data, String imei) {
        //6点到10点之间的定位信息走队列解析
        //if("location".equals(kafkaDataType) && TimeUtils.isBetweenMAndN(6,10)) {
        if("location".equals(kafkaDataType)) {
              data.put("imei", imei);
              data.put("useWeiZhiLocateBL",useWeiZhiLocateBL_LBS);
              boolean isSuccess = LimitedSingleQueue.getInstance().submit(data);
              //队列满了直接消费
              if(!isSuccess) {
                  //直接解析
                  analysisKafkaData(data, imei);
              } else {
                  log.info("定位信息成功加入了消费队列......");
              }
        } else {
            //直接解析
            analysisKafkaData(data, imei);
        }
    }

    /**
     * @description 解析数据
     * @author cwj
     * @param [data, imei, kafkaModelType]
     * @return void
     * @date @time 2025/6/4 15:30
     **/
    private void analysisKafkaData(JSONObject data, String imei) {
        tcpService.useWeiZhiLocateBL = Double.valueOf(useWeiZhiLocateBL_LBS);
        tcpService.cmdTempFunc(imei, data);
    }


    /**
     * @description 消费Kafka数据
     * @author cwj
     * @param [messages]
     * @return void
     * @date @time 2025/5/26 11:16
     **/
    @KafkaListener(groupId = "${spring.kafka.consumer.group.id-wechatVedio}", topics = "${spring.kafka.consumer.topic.name-wehcatVedio}", concurrency = "${spring.kafka.consumer.concurrency-wechatVedio}")
    public void listenWechatVedio(List<ConsumerRecord<String, String>> messages){
        if(Integer.valueOf(1).equals(isCloseTcpPushWechatVedioConsumer)) {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        log.info("topic为pinxuan_TCPPUSH_Wechat_Vedio消息队列接收到的消息为:{}",message);
                    }
                }
            }
        } else {
            if(messages != null){
                for (ConsumerRecord<String, String> message: messages){
                    if(message != null){
                        String imei = message.key();
                        String value = message.value().replace("\r", "").replace("\n", "")
                                .replace("\ufeff", "");
                        log.info("topic={}. imei={}. value={}. partition={}", message.topic(), imei, value, message.partition());
                        try{
                            JSONObject data = JSON.parseObject(value);
                            if(data != null && StringUtil.isNotBlank(imei)){
                                tcpService.useWeiZhiLocateBL = Double.valueOf(useWeiZhiLocateBL_LBS);
                                //kafka消费模式: 0:普通 1:LBS 2:VIP，3：微聊或视频通话
                                data.put("kafkaModelType", 3);
                                tcpService.cmdTempFunc(imei, data);
                            }
                        } catch (Exception e){
                            log.error(String.format("微聊、视频通话消费主题 Exception = %s. imei = %s. message = %s", e.getMessage(), imei, message), e);
                        }
                    } else {
                        log.error("listenWechatVedio message is null.");
                    }
                }
            } else {
                log.error("listenWechatVedio messages is null.");
            }
        }
    }
}
